package com.group1.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.group1.realtime.app.BaseAppV1;
import com.group1.realtime.bean.BroadCastVideoTimesBean;
import com.group1.realtime.common.Constant;
import com.group1.realtime.function.DimAsyncFunction;
import com.group1.realtime.util.AtguiguUtil;
import com.group1.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class Dws_BroadCast_VideoTimes extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_BroadCast_VideoTimes().init(
                3088,
                2,
                "Dws_BroadCast_VideoTimes",
                Constant.TOPIC_DWD_TRAFFIC_APPVIDEO
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          DataStreamSource<String> stream) {

        // 1. 按照video_id进程去重
        SingleOutputStreamOperator<JSONObject> distinctStream = distinctByViedoId(stream);
        // 2. 把数据封装到pojo中
        SingleOutputStreamOperator<BroadCastVideoTimesBean> beanStream = parseToPojo(distinctStream);
        // 3. 补充维度信息
        SingleOutputStreamOperator<BroadCastVideoTimesBean> beanStreamWithDim = addDim(beanStream);
        //beanStreamWithDim.print();
        // 4. 按照chapter_id分组, 开窗聚合
        SingleOutputStreamOperator<BroadCastVideoTimesBean> resultStream = windowAndAgg(beanStreamWithDim);
        // 5. 数据写出到clickhouse中
        resultStream.print();
        //writeToClickHouse(resultStream);

    }

    private void writeToClickHouse(SingleOutputStreamOperator<BroadCastVideoTimesBean> resultStream) {

        resultStream.addSink(FlinkSinkUtil.getClickHouseSink("", BroadCastVideoTimesBean.class));
    }

    private SingleOutputStreamOperator<BroadCastVideoTimesBean> windowAndAgg(
            SingleOutputStreamOperator<BroadCastVideoTimesBean> beanStreamWithDim) {

        return beanStreamWithDim
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<BroadCastVideoTimesBean>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                                .withIdleness(Duration.ofSeconds(1))
                )
                .keyBy(BroadCastVideoTimesBean::getChapter_id)
                .window(TumblingEventTimeWindows.of(Time.seconds(2)))
                .reduce(new ReduceFunction<BroadCastVideoTimesBean>() {
                            @Override
                            public BroadCastVideoTimesBean reduce(BroadCastVideoTimesBean value1,
                                                                  BroadCastVideoTimesBean value2) throws Exception {
                                value1.setBroadCastCount(value1.getBroadCastCount() + value2.getBroadCastCount());
                                return value1;
                            }
                        },
                        new ProcessWindowFunction<BroadCastVideoTimesBean, BroadCastVideoTimesBean, String, TimeWindow>() {
                            @Override
                            public void process(String key,
                                                Context ctx,
                                                Iterable<BroadCastVideoTimesBean> elements,
                                                Collector<BroadCastVideoTimesBean> out) throws Exception {

                                BroadCastVideoTimesBean bean = elements.iterator().next();
                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));

                                bean.setTs(ctx.currentProcessingTime());

                                out.collect(bean);
                            }
                        }
                );
    }

    private SingleOutputStreamOperator<BroadCastVideoTimesBean> addDim(SingleOutputStreamOperator<BroadCastVideoTimesBean> beanStream) {

        return AsyncDataStream.unorderedWait(
                beanStream,
                new DimAsyncFunction<BroadCastVideoTimesBean>() {

                    @Override
                    public String getTable() {
                        return "dim_video_info";
                    }

                    @Override
                    public String getId(BroadCastVideoTimesBean input) {
                        return input.getVideo_id();
                    }

                    @Override
                    public void addDim(BroadCastVideoTimesBean input,
                                       JSONObject dim) {
                        input.setChapter_id(dim.getString("CHAPTER_ID"));
                    }
                },
                60,   // 异步处理超时时间
                TimeUnit.SECONDS  // 超时赶时间的单位
        );
    }

    private SingleOutputStreamOperator<BroadCastVideoTimesBean> parseToPojo(SingleOutputStreamOperator<JSONObject> distinctStream) {

        return distinctStream
                .map(object -> {
                    JSONObject obj = JSON.parseObject(String.valueOf(object));

                    return BroadCastVideoTimesBean.builder()
                            .video_id(obj.getJSONObject("appVideo").getString("video_id"))
                            .broadCastCount(1L)
                            .ts(obj.getLong("ts"))
                            .build();
                });
    }

    private SingleOutputStreamOperator<JSONObject> distinctByViedoId(DataStreamSource<String> stream) {

        return stream
                .map(JSON::parseObject)
                .keyBy(object -> object.getJSONObject("appVideo").getString("video_id"))
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    private ValueState<JSONObject> singleVideoIdState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        singleVideoIdState = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("singleVideoIdState", JSONObject.class));
                    }

                    @Override
                    public void processElement(JSONObject value,
                                               Context ctx,
                                               Collector<JSONObject> out) throws Exception {

                        //第一条数据进来的时候, 状态应该是空, 注册定时器, 更新状态
                        if (singleVideoIdState.value() == null) {
                            // 1. 更新状态
                            singleVideoIdState.update(value);

                            out.collect(singleVideoIdState.value());
                        }

                    }
                });
    }
}
